package com.bjpowernode.shop.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.bjpowernode.shop.base.BaseProduct;
import com.bjpowernode.shop.constants.QueueConstants;
import com.bjpowernode.shop.domain.*;
import com.bjpowernode.shop.pool.ProductThreadPool;
import com.bjpowernode.shop.repository.ProdEsRepository;
import com.bjpowernode.shop.service.ImportService;
import com.rabbitmq.client.Channel;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

@Service
public class ImportServiceImpl extends BaseProduct implements ImportService, CommandLineRunner {

    @Value("${es.import.size}")
    private Integer size;

    @Autowired
    private ProdEsRepository prodEsRepository;

    private Date t1;

    /*
        全量导入操作，在启动微服务时，将数据库中的数据全部查询出来
        并且完成转换 List<Prod> -> List<ProdEs>
        批量导入操作
     */
    @Override
    // ====分页查询 = 查询总记录数 + 分页查询====
    public void importAll() {
        importProd2Es(null, null);
    }

    private void importProd2Es(Date t1, Date t2) {
        System.out.println("##########");

        // 查询总记录数
        Integer totalCounts = prodMapper.selectCount(
                new LambdaQueryWrapper<Prod>().eq(
                        Prod::getStatus, 1
                ).between(t1 != null && t2 != null, Prod::getUpdateTime, t1, t2)
        );

        // 总页数 = 总记录数 % 每页条数 == 0 ? 总记录数 / 每页条数 : (总记录数 / 每页条数) + 1
        Integer totalPages = totalCounts % size == 0 ? totalCounts / size : (totalCounts / size) + 1;

        // ===通过计数器控制t1赋值时机===
        CountDownLatch countDownLatch = new CountDownLatch(totalPages);

        // ==遍历总页数，分批次的导入数据到es中==
        for (int i = 0; i < totalPages; i++) {
            // 获取分页数据
            Page<Prod> prodPage = prodMapper.selectPage(
                    // 分页查询，MybatisPlus的第一页从1开始的，0和1查询的都是第一页数据
                    new Page<Prod>(i + 1, size),
                    new LambdaQueryWrapper<Prod>().eq(
                            Prod::getStatus, 1
                    ).between(t1 != null && t2 != null, Prod::getUpdateTime, t1, t2)
            );
            System.out.println("prodPage = " + prodPage);

            // 从分页中获取数据
            List<Prod> prodList = prodPage.getRecords();
            if (CollectionUtils.isEmpty(prodList))
                return;

            // 数据转换，List<Prod> -> List<ProdEs>
            List<ProdEs> prodEsList = changeObjList(prodList);
            System.out.println("prodEsList = " + prodEsList);

            // 将数据导入es
            // 通过线程池，来维护耗时操作，也不用反复创建线程，消耗性能资源
            ProductThreadPool.poolExecutor.execute(
                    () -> prodEsRepository.saveAll(prodEsList)
            );

            // ===计数器=== 导入完成一次，计数器减一
            countDownLatch.countDown();
        }

        // 在全部导入完成前，都会堵塞在此处
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    private List<ProdEs> changeObjList(List<Prod> prodList) {
        List<ProdEs> prodEsList = new ArrayList<>();
        // 拷贝实体类
        prodList.forEach(prod -> {
            ProdEs prodEs = new ProdEs();
            BeanUtils.copyProperties(prod, prodEs);
            prodEsList.add(prodEs);
        });
        System.out.println("changeObjList.prodEsList = " + prodEsList);

        prodList.forEach(

                prod -> {
                    // 查询当前产品的tag中间表对象的集合
                    List<ProdTagReference> prodTagReferences = prodTagReferenceMapper.selectList(
                            new LambdaQueryWrapper<ProdTagReference>()
                                    .eq(
                                            ObjectUtils.isNotEmpty(prod.getProdId()),
                                            ProdTagReference::getProdId,
                                            prod.getProdId()
                                    ));
                    // 取出集合中所有对象的tagId作为tagIdList
                    List<Long> tagIdList = prodTagReferences.stream().map(ProdTagReference::getTagId).collect(Collectors.toList());

                    // 查询好评数
                    Integer praiseNumber = prodCommMapper.selectCount(new LambdaQueryWrapper<ProdComm>()
                            .eq(ObjectUtils.isNotEmpty(prod.getProdId()), ProdComm::getProdId, prod.getProdId())
                            .eq(ObjectUtils.isNotEmpty(prod.getProdId()), ProdComm::getEvaluate, 0)
                    );

                    // 查询评论总数
                    Integer totalComm = prodCommMapper.selectCount(new LambdaQueryWrapper<ProdComm>()
                            .eq(ObjectUtils.isNotEmpty(prod.getProdId()), ProdComm::getProdId, prod.getProdId())
                            .eq(ObjectUtils.isNotEmpty(prod.getProdId()), ProdComm::getStatus, 1)
                    );

                    // 计算好评率
                    BigDecimal positiveRating;
                    if (totalComm.intValue() > 0) {
                        positiveRating = new BigDecimal(praiseNumber)
                                .divide(new BigDecimal(totalComm), 2, RoundingMode.HALF_UP)
                                .multiply(new BigDecimal(100));
                    } else {
                        positiveRating = BigDecimal.ZERO;
                    }

                    List<ProdEs> collect = prodEsList.stream().filter(
                            prodEs -> prodEs.getProdId().equals(prod.getProdId())
                    ).collect(Collectors.toList());
                    ProdEs currProdEs = collect.get(0);

                    currProdEs.setTagList(tagIdList) // 产品标签ids的列表
                            .setPraiseNumber(praiseNumber.longValue()) // 好评数
                            .setPositiveRating(positiveRating); // 好评率
                }
        );

        return prodEsList;
    }

    @Override
    // initialDelay 第一执行的时间
    // fixedDelay 后续每次执行的时间
    // @Scheduled(initialDelay = 1000 * 30, fixedDelay = 1000 * 30)
    public void updateImport() {
        Date t2 = new Date();
        importProd2Es(t1, t2);
        System.out.println("***t2 = " + t2);
        t1 = t2;
        System.out.println("***New t1 = " + t1);
    }

    @Override
    @RabbitListener(queues = {QueueConstants.ES_CHANGE_QUEUE})
    public void quickImport(Message message, Channel channel) {
        
        // 获取消息对象：[{prodId:94,count:1},{prodId:95,count:2},{prodId:96,count:3}]
        String json = new String(message.getBody());
        List<ProdCount> prodCounts = JSON.parseArray(json, ProdCount.class);

        // 要导入的结果集
        List<ProdEs> prodEsList = new ArrayList<>();
        
        // 遍历集合中的对象
        prodCounts.forEach(
                prodCount -> {
                    // 获取与当前对象对应的es数据
                    Optional<ProdEs> optionalProdEs = prodEsRepository.findById(prodCount.getProdId());
                    ProdEs prodEs = optionalProdEs.get();
                    // 更新es数据的count
                    Integer newStocks = prodEs.getTotalStocks() + prodCount.getCount().intValue();
                    prodEs.setTotalStocks(newStocks);
                    // es数据加入结果集中备用
                    prodEsList.add(prodEs);
                    prodEsRepository.save(prodEs);
                }
        );
        
        // prodEsRepository.saveAll(prodEsList);

        try {
            channel.basicAck(
                    message.getMessageProperties().getDeliveryTag(),false
            );
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    @Override
    public void run(String... args) throws Exception {
        // importAll();
        System.out.println(" ************ ");
        t1 = new Date();
        System.out.println("***t1 = " + t1);
        // prodEsRepository.findAll().forEach(System.out::println); // 查看tagList
    }
}
